package cn.xinfei.xdecision.data.core.frame.executors.collect.handler;

import cn.xinfei.xdecision.common.model.datax.consts.VariableConst;
import cn.xinfei.xdecision.common.model.datax.datainterface.v2.VarInterfaceInfo;
import cn.xinfei.xdecision.common.model.datax.datamanage.v2.Variable;
import cn.xinfei.xdecision.common.model.datax.enums.EnumUtil;
import cn.xinfei.xdecision.common.model.datax.enums.ExecutorEnum;
import cn.xinfei.xdecision.common.model.datax.exception.XDecisionDataException;
import cn.xinfei.xdecision.common.model.datax.exception.XDecisionDataExceptionType;

import cn.xinfei.xdecision.common.service.datax.datainterface.v2.VarInterfaceInfoService;
import cn.xinfei.xdecision.common.service.datax.datamanage.v2.VariableService;
import cn.xinfei.xdecision.data.core.cache.VarInterfaceCache;
import cn.xinfei.xdecision.data.core.cache.VarMetaDataCache;
import cn.xinfei.xdecision.data.core.frame.executors.collect.VarHandler;
import cn.xinfei.xdecision.data.core.frame.executors.VarContext;
import cn.xinfei.xdecision.data.core.frame.executors.collect.filter.FilterManage;
import cn.xinfei.xdecision.data.core.frame.executors.collect.filter.GroovyAfterVarFilter;
import cn.xinfei.xdecision.data.core.frame.executors.collect.filter.GroovyPreVarFilter;
import cn.xinfei.xdecision.data.core.vo.VarCollectAwareVO;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.MapDifference;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.stream.Collectors;

@Service
@Slf4j
public class VarRuteDataSourceHandler implements VarHandler<VarCollectAwareVO> {
    @Autowired
    VarInterfaceInfoService varInterfaceInfoService;

    @Autowired
    VariableService variableService;

    @Autowired
    VarMetaDataCache varMetaDataCache;

    @Autowired
    VarInterfaceCache varInterfaceCache;


    @Override
    public void doHandle(VarContext<VarCollectAwareVO> context) {
        VarCollectAwareVO varCollectAwareVO = context.get();
        Set<String> varCodes = varCollectAwareVO.getVarCodes();
        String requestId = varCollectAwareVO.getRequestId();
        Map<String, Set<String>> derivedVarAndParamMap = varCollectAwareVO.getDerivedVarDependParamMap();
        Map<String, Integer> varCodeAndVarTypeMap = varCollectAwareVO.getVarCodeAndTypeMap();

        Optional<Map<String, Variable>> varsMetaDataCache = Optional.ofNullable(varMetaDataCache.loadVarMetaData());
        Optional<Map<Integer, VarInterfaceInfo>> varInterfaceInfoCache = Optional.ofNullable(varInterfaceCache.loadVarInterfaceMetaData());
        List<String> usedExecutors;

        if (varsMetaDataCache.isPresent() && varInterfaceInfoCache.isPresent()) {
            log.info("requestId = {} VarRuteDataSourceHandler 查询缓存 VarMetaDataCache & VarInterfaceInfoCache ", requestId);
            usedExecutors = findExecutorsFromCacheToRun(varCodes, varsMetaDataCache.get(), varInterfaceInfoCache.get());
        } else {
            usedExecutors = findExecutorsToRun(varCodes);
        }
        Set<String> varCodesNeeded = buildVarCodesNeeded(varCollectAwareVO);

        Map<Integer, Set<String>> varTypeAndVarCodesMap;
        if (varsMetaDataCache.isPresent()) {
            log.info("requestId = {} VarRuteDataSourceHandler 查询缓存 VarMetaDataCache ", requestId);
            varTypeAndVarCodesMap = variableService.findVarTypeAndVarCodesMapFromCache(varCodesNeeded, varsMetaDataCache.get());
        } else {
            varTypeAndVarCodesMap = variableService.findVarTypeAndVarCodesMap(varCodesNeeded);
        }

        switch (varCollectAwareVO.getCollectStageEnum()) {
            case PRE_COLLECT:
                Map<String, Set<String>> executorAndFreeVarCodesMap = Maps.newHashMap();
                Map<String, Set<String>> finalExecutorAndVarCodesMap = Maps.newHashMap();

                if (MapUtils.isNotEmpty(varTypeAndVarCodesMap)) {
                    executorAndFreeVarCodesMap = varTypeAndVarCodesMap.entrySet().stream()
                            .filter(item -> VariableConst.PRE_COLLECT_VAR_TYPE.contains(item.getKey()))
                            .collect(Collectors.toMap(item -> VariableConst.VAR_AND_EXECUTOR_TYPE_MAP.get(item.getKey()), Map.Entry::getValue));

                }
                Set<String> preCollectExecutorVars = executorAndFreeVarCodesMap.get(VariableConst.PRE_COLLECT_EXECUTOR_ENUMS.get(0));

                Set<String> exeVarcodes = executorAndFreeVarCodesMap.get(VariableConst.PRE_COLLECT_EXECUTOR_ENUMS.get(0));
                if (CollectionUtils.isNotEmpty(exeVarcodes)) {
                    finalExecutorAndVarCodesMap = new HashMap<String, Set<String>>() {
                        {
                            put(VariableConst.PRE_COLLECT_EXECUTOR_ENUMS.get(0), exeVarcodes);
                        }
                    };
                }

                varCollectAwareVO.setVarCodes(preCollectExecutorVars);
                varCollectAwareVO.setExecutorAndVarCodesMap(finalExecutorAndVarCodesMap);
                varCollectAwareVO.setUsedExecutorEnums(usedExecutors);
                context.setContext(varCollectAwareVO);
                log.info("requestId = {}  RutingDataHandler executorVars:{} ", requestId, JSONObject.toJSONString(preCollectExecutorVars));
                break;

            case REAL_TIME_COLLECT:
                Map<String, Set<String>> executorAndVarCodesMap = Maps.newHashMap();
                Set<String> realTimeExecutorVars = new HashSet<>();

                if (MapUtils.isNotEmpty(varTypeAndVarCodesMap)) {
                    executorAndVarCodesMap = varTypeAndVarCodesMap.entrySet().stream()
                            .filter(item -> VariableConst.REAL_COLLECT_VAR_TYPE.contains(item.getKey()))
                            .collect(Collectors.toConcurrentMap(item -> VariableConst.VAR_AND_EXECUTOR_TYPE_MAP.get(item.getKey()), Map.Entry::getValue));
                    if (MapUtils.isNotEmpty(executorAndVarCodesMap)) {
                        realTimeExecutorVars = executorAndVarCodesMap.values().stream().flatMap(Collection::stream).filter(Objects::nonNull).collect(Collectors.toSet());
                    }
                }

                Set<String> groovyPreVars = FilterManage.vars(varCodes)
                        .addFilter(new GroovyPreVarFilter(derivedVarAndParamMap, varCodeAndVarTypeMap))
                        .invokerFilter(String.class);

                Map<String, Set<String>> groovyPreVarMap = derivedVarAndParamMap.entrySet().stream()
                        .filter(item -> groovyPreVars.contains(item.getKey()))
                        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));


                Set<String> groovyAfterVars = FilterManage.vars(varCodes)
                        .addFilter(new GroovyAfterVarFilter(derivedVarAndParamMap, groovyPreVarMap))
                        .invokerFilter(String.class);

//                Map<String, List<String>> groovyAfterVarMap = derivedVarAndParamMap.entrySet().stream()
//                        .filter(item -> groovyAfterVars.contains(item.getKey()))
//                        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

                MapDifference<String, Set<String>> difference = Maps.difference(derivedVarAndParamMap, groovyPreVarMap);
                Map<String, Set<String>> groovyAfterVarMap = difference.entriesOnlyOnLeft();

                Map<String, Set<String>> executorAndGroovyPreVarCodesMap = constructExecutorAndGroovyVarMap(
                        groovyPreVarMap,
                        ExecutorEnum.GROOVY_PRE
                );
                Map<String, Set<String>> executorAndGroovyAfterVarCodesMap = constructExecutorAndGroovyVarMap(
                        groovyAfterVarMap,
                        ExecutorEnum.GROOVY_AFTER
                );

                log.info("requestId = {} executorAndGroovyPreVarCodesMap:{}, executorAndGroovyAfterVarCodesMap:{}, groovyPreVarMap:{}, groovyAfterVarMap:{}",
                        requestId,
                        JSONObject.toJSONString(executorAndGroovyPreVarCodesMap),
                        JSONObject.toJSONString(executorAndGroovyAfterVarCodesMap),
                        JSONObject.toJSONString(groovyPreVarMap),
                        JSONObject.toJSONString(groovyAfterVarMap));


                executorAndVarCodesMap.putAll(executorAndGroovyPreVarCodesMap);
                executorAndVarCodesMap.putAll(executorAndGroovyAfterVarCodesMap);

                List<String> externalVar = fetchGroovyExternalVarToExec(executorAndVarCodesMap);
                realTimeExecutorVars.addAll(externalVar);
                usedExecutors = executorAndVarCodesMap.entrySet().stream().map(Map.Entry::getKey).collect(Collectors.toList());

                varCollectAwareVO.setVarCodes(realTimeExecutorVars);
                varCollectAwareVO.setExecutorAndVarCodesMap(executorAndVarCodesMap);
                varCollectAwareVO.setUsedExecutorEnums(usedExecutors);
                context.setContext(varCollectAwareVO);
                log.info("requestId = {}  RutingDataHandler executorVars:{} usedExecutors:{} ",
                        requestId,
                        JSONObject.toJSONString(realTimeExecutorVars),
                        JSONObject.toJSONString(usedExecutors));
                break;
            default:
                throw new XDecisionDataException(XDecisionDataExceptionType.DATA_NO_SUCH_COLLECT_STAGE_ERROR, "没有该类型的采集: " + context.get().getCollectStageEnum());
        }
    }


    private List<String> fetchGroovyExternalVarToExec(Map<String, Set<String>> executorAndVarCodesMap) {
        if (MapUtils.isEmpty(executorAndVarCodesMap)) {
            return Collections.emptyList();
        }
        return executorAndVarCodesMap.entrySet().stream()
                .filter(executor -> VariableConst.GROOVY_EXECUTOR_ENUMS.contains(executor.getKey()))
                .collect(HashMap::new, (map, entry) -> map.put(entry.getKey(), entry.getValue()), HashMap::putAll)
                .values().stream()
                .filter(Objects::nonNull)
                .map(String::valueOf)
                .distinct().collect(Collectors.toList());

    }

    private Map<String, Set<String>> constructExecutorAndGroovyVarMap(
            Map<String, Set<String>> groovyVarMap,
            ExecutorEnum executorEnum) {
        Map<String, Set<String>> executorAndVarsMap = Maps.newHashMap();
        if (MapUtils.isNotEmpty(groovyVarMap)) {
            Set<String> groovyVars = new HashSet<>(groovyVarMap.keySet());
            switch (executorEnum) {
                case GROOVY_PRE:
                    executorAndVarsMap.put(executorEnum.name(), groovyVars);
                    break;
                case GROOVY_AFTER:
                    executorAndVarsMap.put(executorEnum.name(), groovyVars);
                    break;
                default:
                    throw new XDecisionDataException(XDecisionDataExceptionType.DATA_NO_SUCH_EXECUTOR_ERROR, "没有该类型的执行器: " + executorEnum.getDesc());
            }

        }
        return executorAndVarsMap;

    }

    private Set<String> buildVarCodesNeeded(VarCollectAwareVO varCollectAwareVO) {
        return varCollectAwareVO.getApiVariablesMap().values().stream().flatMap(Collection::stream).collect(Collectors.toSet());
    }


    private List<String> findExecutorsToRun(Collection<String> varCodes) {
        List<Long> interfaceIds = variableService.findDistinctInterfaceIdsByVarCodes(varCodes);

        if (CollectionUtils.isEmpty(interfaceIds)) {
            return Collections.emptyList();
        }
        return varInterfaceInfoService.selectTypesByIds(interfaceIds).stream()
                .map(VariableConst.DATASOURCE_AND_EXECUTOR_TYPE_MAP::get)
                .map(item ->
                        EnumUtil.fromCode(ExecutorEnum.class, item)
                ).filter(Objects::nonNull).map(Enum::name).distinct().collect(Collectors.toList());
    }

    private List<String> findExecutorsFromCacheToRun(Collection<String> varCodes,
                                                     Map<String, Variable> variableCache,
                                                     Map<Integer, VarInterfaceInfo> varInterFaceCache) {

        List<Long> interfaceIds = variableCache.values()
                .stream()
                .filter(var -> varCodes.contains(var.getVarCode()))
                .map(Variable::getInterfaceId)
                .collect(Collectors.toList());

        if (CollectionUtils.isEmpty(interfaceIds)) {
            return Collections.emptyList();
        }
        return varInterFaceCache.values()
                .stream()
                .filter(varInterface ->
                        interfaceIds.contains(Long.valueOf(varInterface.getId())) &&
                                varInterface.getStatus() == 1
                )
                .map(VariableConst.DATASOURCE_AND_EXECUTOR_TYPE_MAP::get)
                .map(item ->
                        EnumUtil.fromCode(ExecutorEnum.class, item)
                ).filter(Objects::nonNull).map(Enum::name).distinct().collect(Collectors.toList());
    }

}
